package org.apache.flink.streaming.api.datastream;

import iscas.RedisBarrier;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

public class ABSBarrier<D extends ABSMarkable> extends RichFlatMapFunction<D, D> {
    ArrayList<D> buffer;
    boolean mark;

    @Override
    public void open(Configuration parameters) {
        buffer = new ArrayList<>(100);
    }

    @Override
    public void flatMap(D in, Collector<D> out) {
        if (RedisBarrier.isBarrier()) {
            if (mark) {
                if (buffer.size() > 100) {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException ignored) {
                    }
                }
                buffer.add(in);
            } else {
                in.mark();
                out.collect(in);
                mark = true;
            }
        } else {
            if (!buffer.isEmpty()) {
                buffer.forEach(out::collect);
                System.out.printf("[ISCAS][ABS]buffer count: %d\n", buffer.size());
                buffer.clear();
            }
            mark = false;
            out.collect(in);
        }
    }
}
